package com.myrabbitmqdemo.demo.four;

import com.myrabbitmqdemo.demo.utils.RabbitMqUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.DeliverCallback;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.TimeoutException;

/**
 * 生产者
 * 发布确认模式：
 *  1、单个确认模式
 *  2、批量确认模式
 *  3、异步批量确认
 */

public class ConfirmMessage {

    private static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws InterruptedException, TimeoutException, IOException {
        //        1、单个确认模式
        //        2、批量确认模式
        //        3、异步批量确认

        //  ConfirmMessage.publishMessageIndividually();     //  完成1000条单个确认模式花费18238ms
        ConfirmMessage.publishMessageAsync();
    }

    // 单个确认
    public static void publishMessageIndividually() throws IOException, TimeoutException, InterruptedException {
        Channel channel = RabbitMqUtils.getChannel();

        String queueName = UUID.randomUUID().toString();

        channel.queueDeclare(queueName,true,false,false,null);

        // 开启发布确认模式
        channel.confirmSelect();

        long begin = System.currentTimeMillis();

        for (int i=0;i<MESSAGE_COUNT;i++) {
            String message = i + "";
            channel.basicPublish("",queueName,null,message.getBytes(StandardCharsets.UTF_8));

            boolean b = channel.waitForConfirms();

            if (b) {
                System.out.println("发布已完成:"+i);
            }
        }
        long end = System.currentTimeMillis();

        System.out.println("完成1000条单个确认模式花费"+(end-begin)+"ms");

    }

    // 异步批量确认发布
    public static void publishMessageAsync() throws IOException, TimeoutException {
        Channel channel = RabbitMqUtils.getChannel();

        String queueName = UUID.randomUUID().toString();

        channel.queueDeclare(queueName,true,false,false,null);

        channel.confirmSelect();

        long begin = System.currentTimeMillis();

        /**
         * 消息成功的回调
         */
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            // 2、删除掉已经确认的消息，剩下的就是未确认的消息
            System.out.println("确认的消息：" + deliveryTag);
        };

        /**
         * 消息失败的回调：
         * 1、消息得标记
         * 2、是否是批量确认
         */
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            // 3、打印一下未确认的消息都有哪些
            System.out.println("未得到确认得消息：" + deliveryTag);
        };

        // 开始准备监听器，监听哪些消息成功了，哪些消息失败了
        channel.addConfirmListener(ackCallback,nackCallback);

        for (int i=0;i<MESSAGE_COUNT;i++) {
            String message = "消息：" + i;
            channel.basicPublish("",queueName,null,message.getBytes(StandardCharsets.UTF_8));

            // 1、此处记录下所有的要发送的消息，消息的总和
        }

        long end = System.currentTimeMillis();

        System.out.println("消息队列多线程总共消耗了:"+ (end-begin) + "ms");

    }
}
